package org.example.seatunnel;

import cn.hutool.core.util.StrUtil;
import com.typesafe.config.Config;
import org.example.SqlTablePrefixer;
import org.example.datasource.DataSourceApi;
import org.example.datasource.impl.DataLinkMedataDataSourceImpl;
import org.example.datasource.model.JdbcDataSourceInfo;
import org.example.datasource.model.TableMetadata;
import org.example.seatunnel.model.SinkResModel;
import org.example.seatunnel.model.SourceResModel;

import java.util.List;

public class SeatunnelJdbcSink implements SeatunnelSink {


    @Override
    public SinkResModel parse(SourceResModel sourceResModel , Config config) {
        if (config.hasPath("Jdbc")){
            config = config.getConfig("Jdbc");
        }
        if (config.hasPath("jdbc")){
            config = config.getConfig("jdbc");

        }
        JdbcDataSourceInfo jdbcDataSourceInfo = new JdbcDataSourceInfo();
        jdbcDataSourceInfo.setUrl(config.getString("url"));
        jdbcDataSourceInfo.setPassword(config.getString("password"));
        jdbcDataSourceInfo.setUsername(config.getString("user"));

        DataSourceApi dataSourceApi = new DataLinkMedataDataSourceImpl(jdbcDataSourceInfo);


        try{
            if (config.hasPath("query")){
                SinkResModel sourceResModelRes = new SinkResModel();
                sourceResModelRes.setConfig(config);

                String sql =  config.getString("query");
                sql = sql.replaceAll("\\[","");
                sql = sql.replaceAll("]","");
                sql = sql.replaceAll("`","");

                // 截取 insert into sql
                sourceResModelRes.setSql( modifyInsertStatement(sql,"sink.")
                        +"  "+ SqlTablePrefixer.addPrefixToTables(sourceResModel.getSql(),"source."));


                List<TableMetadata> tableMetadataList =  dataSourceApi.getTableMetadata(sql,true);

//                System.out.println("获取sink: " + jdbcDataSourceInfo.getUrl() +"  元数据:" +tableMetadataList.size());
//                System.out.println(List.of(getSinkTableName(config.getString("query"))));

                sourceResModelRes.setSinkMetadataList(tableMetadataList);
                sourceResModelRes.setSourceMetadataList(sourceResModel.getTableMetadataList());
                return sourceResModelRes;
            }
            if (config.hasPath("table")){
                SinkResModel sourceResModelRes = new SinkResModel();
                sourceResModelRes.setConfig(config);
                sourceResModelRes.setSql(StrUtil.format(" insert into sink.{} {}", config.getString("table"),
                        SqlTablePrefixer.addPrefixToTables(sourceResModel.getSql(),"source.")));
                sourceResModelRes.setSourceMetadataList(sourceResModel.getTableMetadataList());
                List<TableMetadata> tableMetadataList =  dataSourceApi.getTableMetadata(List.of(config.getString("table")));
//                System.out.println("获取sink: " + jdbcDataSourceInfo.getUrl() +"  元数据:" +tableMetadataList.size());
//                System.out.println(List.of(config.getString("table")));

                sourceResModelRes.setSourceMetadataList(sourceResModel.getTableMetadataList());

                sourceResModelRes.setSinkMetadataList(tableMetadataList);

                return sourceResModelRes;

            }
        }catch (Exception e){
            throw e;
        }

        return null;
    }
    public static String modifyInsertStatement(String sql, String prefix) {
        // 查找 INSERT INTO 之后的部分
        int startIndex = sql.toUpperCase().indexOf("INSERT INTO") + "INSERT INTO".length();
        int endIndex = sql.indexOf("(", startIndex);

        if (startIndex != -1 && endIndex != -1) {
            // 提取表名
            String tableName = sql.substring(startIndex, endIndex).trim();
            // 替换原表名为带前缀的表名
            String newTableName = prefix + tableName;
            // 构造新的SQL语句
            sql = sql.substring(0, startIndex) + " " + newTableName + sql.substring(endIndex);
        }

        return extractInsertStatement(sql);
    }

    public static String getSinkTableName(String sql) {
        // 查找 INSERT INTO 之后的部分
        int startIndex = sql.toUpperCase().indexOf("INSERT INTO") + "INSERT INTO".length();
        int endIndex = sql.indexOf("(", startIndex);

        if (startIndex != -1 && endIndex != -1) {
            // 提取表名
            String tableName = sql.substring(startIndex, endIndex).trim();
            // 替换原表名为带前缀的表名
            // 构造新的SQL语句
            return tableName;
        }
        return null;
    }

    public static String extractInsertStatement(String sql) {
        // 查找 VALUES 关键字的位置
        int index = sql.toUpperCase().indexOf(" VALUES");
        if (index != -1) {
            // 截取 VALUES 之前的部分
            return sql.substring(0, index);
        } else {
            // 如果没有找到 VALUES，返回原始字符串
            return sql;
        }
    }

    @Override
    public boolean isSupport(Config config) {
        if (config.hasPath("Jdbc")){
            return true;
        }
        if (config.hasPath("jdbc")){
            return true;
        }
        return false;
    }
}
